package com.bjbn.aesbcommonapi.service;

import com.alibaba.fastjson.JSONObject;
import com.bjbn.aesbcommonapi.utils.MsgFormatUtil;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
@RequiredArgsConstructor
public class MessageTransferService {
    private static final Logger logger = LoggerFactory.getLogger(MessageTransferService.class);

    private final KafkaTemplate<String, String> kafkaTemplate;

    // 标准格式主题
    private static final String STANDARD_TOPIC = "T-CAN-JSON-ORI";

    public void transfer(String resourceTopic, MessageExt message) {
        try {
            String payload = new String(message.getBody());

            // 解析消息体判断格式
            JSONObject messageJson = JSONObject.parseObject(payload);

            // 判断是否为标准格式（包含producer、from、bizId、time、data字段）
            if (MsgFormatUtil.isStandardFormat(messageJson)) {
                // 标准格式推送到T-CAN-JSON-ORI主题
                kafkaTemplate.send(STANDARD_TOPIC, payload);
                logger.debug("转发标准格式消息到Kafka topic: {}，消息ID: {}，消息详情：{}", STANDARD_TOPIC, message.getMsgId(), payload);
            } else if ("T-CAN-IOT-DEVICEDATA".equals(resourceTopic)) {
                logger.info("跳过物联网平台格式消息，消息ID: {}，消息详情：{}", message.getMsgId(), payload);
            } else {
                // 其他格式暂不推送
                logger.debug("跳过其他格式消息，消息ID: {}，消息详情：{}", message.getMsgId(), payload);
            }
        } catch (Exception e) {
            logger.error("转发消息到Kafka失败，topic: {}，消息ID: {}", resourceTopic, message.getMsgId(), e);
        }
    }

    public void transferBody(String kafkaTopic, JSONObject obj) {
        try {
            // 物联网数据需要特殊处理
            if ("T-CAN-IOT-DEVICEDATA".equals(kafkaTopic)) {
                // 物联网平台的数据格式有多种，每种格式的数据推送到对应的kafka中
                if (MsgFormatUtil.isIotPlatformFormat(obj)) {
                    kafkaTemplate.send(kafkaTopic, obj.toJSONString());
                } else {
                    /*
                    根据业务数据推送消息体格式，推送到不同的kafka topic中

                    平台产品数据新增/更新/删除格式如下：
                    {
                      "productList": [
                        {
                          "productId": 10033379,
                          "productName": "test009"
                        }
                      ],
                      "messageType": "productOpertaion",
                      "operation": "add",
                      "tenantId": "103***88"
                    }

                    平台设备(元)数据新增/更新/删除格式如下：
                    {
                      "deviceList": [
                        {
                          "deviceId": "2ae27*****0ab42a602114292690",
                          "deviceName": "test009"
                        }
                      ],
                      "messageType": "deviceOperation",
                      "operation": "delete",
                      "productId": 10079,
                      "tenantId": "103***88"
                    }

                    设备档案新增/更新/删除格式如下：
                    {
                      "archiveList": [{
                        "deviceId": "2ae27*****0ab42a602114292690",
                        "deviceName": "test009",
                        "deviceCode": "afafafa",
                        "deviceZone": "ninethZone",
                        "deviceAddress": "秦淮区中山南路**号",
                        "installTime": "",
                        "province": "江苏",
                        "prefecture": "南京",
                        "county": "秦淮"
                      }],
                      "messageType": "archiveOperation",
                      "operation": "add",
                      "productId": 10079,
                      "tenantId": "103***88"
                    }

                    设备数据（上报）数据格式如下：
                    {
                      "upPacketSN": -1,
                      "deviceType": "",
                      "productId": "151***20",
                      "IMEI": "87439*****97353",
                      "deviceId": "5bcba*****f5472fb08a2090cd4e7158",
                      "protocol": "lwm2m",
                      "messageType": "dataReport",
                      "payload": {
                        "current_temperature": -40,
                        "current_humidity": 0
                      },
                      "assocAssetId": "",
                      "tenantId": "9900000000*****",
                      "IMSI": "undefined",
                      "topic": "v1/up/ad19",
                      "upDataSN": -1,
                      "serviceId": 1,
                      "timestamp": 1667800228974
                    }

                    设备事件（告警）格式如下：
                    {
                      "tenantId": "1",
                      "productId": "10010208",
                      "deviceSn": "111qw",
                      "deviceId": "10010208111qw",
                      "messageType": "eventReport",
                      "imei": "xxx",
                      "imsi": "xx",
                      "timestamp": 1554190965950,
                      "eventType": 1,
                      "eventContent": {
                        "m": 44
                      },
                      "serviceId": 1001,
                      "protocol": "rdap"
                    }
                    */
                    // 根据messageType来判断推送kafka topic
                    /*
                    以下代码是cursor生成
                    String messageType = obj.getString("messageType");
                    if ("productOpertaion".equals(messageType) || "productOperation".equals(messageType)) {
                        kafkaTemplate.send("T-CAN-IOT-PRODUCT", obj.toJSONString());
                        logger.debug("转发消息到Kafka topic: {}，消息: {}", "T-CAN-IOT-PRODUCT", obj);
                    } else if ("deviceOperation".equals(messageType)) {
                        kafkaTemplate.send("T-CAN-IOT-DEVICE", obj.toJSONString());
                        logger.debug("转发消息到Kafka topic: {}，消息: {}", "T-CAN-IOT-DEVICE", obj);
                    } else if ("archiveOperation".equals(messageType)) {
                        kafkaTemplate.send("T-CAN-IOT-ARCHIVE", obj.toJSONString());
                        logger.debug("转发消息到Kafka topic: {}，消息: {}", "T-CAN-IOT-ARCHIVE", obj);
                    } else if ("dataReport".equals(messageType)) {
                        kafkaTemplate.send("T-CAN-IOT-UP-PACKET", obj.toJSONString());
                        logger.debug("转发消息到Kafka topic: {}，消息: {}", "T-CAN-IOT-UP-PACKET", obj);
                    } else if ("eventReport".equals(messageType)) {
                        kafkaTemplate.send("T-CAN-IOT-EVENT-REPORT", obj.toJSONString());
                        logger.debug("转发消息到Kafka topic: {}，消息: {}", "T-CAN-IOT-EVENT-REPORT", obj);
                    } else {
                        logger.error("物联网平台数据格式不正确，消息详情：{}", obj);
                    }
                    */
                    // 以下代码是Trae 生成
                    String messageType = obj.getString("messageType");
                    if (messageType != null) {
                        switch (messageType) {
                            case "productOpertaion":
                            case "productOperation":
                                kafkaTemplate.send("T-CAN-IOT-PRODUCT", obj.toJSONString());
                                logger.debug("转发平台产品数据到Kafka topic: T-CAN-IOT-PRODUCT，消息: {}", obj);
                                break;
                            case "deviceOperation":
                                kafkaTemplate.send("T-CAN-IOT-DEVICE", obj.toJSONString());
                                logger.debug("转发平台设备(元)数据到Kafka topic: T-CAN-IOT-DEVICE，消息: {}", obj);
                                break;
                            case "archiveOperation":
                                kafkaTemplate.send("T-CAN-IOT-ARCHIVE", obj.toJSONString());
                                logger.debug("转发设备档案数据到Kafka topic: T-CAN-IOT-ARCHIVE，消息: {}", obj);
                                break;
                            case "dataReport":
                                String productId = obj.getString("productId");
                                // 由于数据量太大，只转发温湿度传感器和光照传感器数据，温湿度传感器productId=10635741，光照传感器productId=10635742
                                if (productId != null) {
                                    if ("10635741".equals(productId) || "10635742".equals(productId)) {
                                        kafkaTemplate.send("T-CAN-IOT-UP-PACKET", obj.toJSONString());
                                        logger.debug("转发设备数据（上报）到Kafka topic: T-CAN-IOT-UP-PACKET，消息: {}", obj);
                                    } else {
                                        logger.debug("跳过非温湿度传感器和光照传感器数据，消息: {}", obj);
                                    }
                                }

                                break;
                            case "eventReport":
                                kafkaTemplate.send("T-CAN-IOT-EVENT-REPORT", obj.toJSONString());
                                logger.debug("转发设备事件（告警）到Kafka topic: T-CAN-IOT-EVENT-REPORT，消息: {}", obj);
                                break;
                            default:
                                logger.debug("跳过未知类型消息，消息类型: {}，消息: {}", messageType, obj);
                                break;
                        }
                    } else {
                        logger.debug("跳过缺少消息类型的消息，消息: {}", obj);
                    }

                }
            } else {
                kafkaTemplate.send(kafkaTopic, obj.toJSONString());
                logger.debug("转发消息到Kafka topic: {}，消息: {}", kafkaTopic, obj);
            }
        } catch (Exception e) {
            logger.error("转发消息到Kafka失败，topic: {}，消息: {}", kafkaTopic, obj, e);
        }
    }


} 